package org.example.statistics

import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}

import java.text.SimpleDateFormat
import java.util.Date

case class Rating(userId: Int, productId: Int, score: Double, timestamp: Int)

case class MongoConfig(uri:String, db:String)

object StatisticsRecommender {

  val MONGODB_RATING_COLLECTION = "Rating"

  //统计的表的名称
  val RATE_MORE_PRODUCTS = "RateMoreProducts" //最近评分
  val RATE_MORE_RECENTLY_PRODUCTS = "RateMoreRecentlyProducts" //最近评分最多的
  val AVERAGE_PRODUCTS = "AverageProducts" //平均数

  // 入口方法
  def main(args: Array[String]): Unit = {

    val config = Map( //定义Map，放相关连接参数
      "spark.cores" -> "local[*]", //本地的核，在windows下指windows，在linux下指linux
      "mongo.uri" -> "mongodb://localhost:27017/recommender", //如果用的是linux下的mongo，则改为linux的ip地址
      "mongo.db" -> "recommender"
    )

    //创建SparkConf配置
    val sparkConf = new SparkConf().setAppName("StatisticsRecommender").setMaster(config("spark.cores")) //本地的核 spark.cores
    //创建SparkSession
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()

    val mongoConfig = MongoConfig(config("mongo.uri"),config("mongo.db"))

    //加入隐式转换
    import spark.implicits._

    //数据加载进来   类似链式编程
    val ratingDF = spark.read
      .option("uri",mongoConfig.uri)
      .option("collection",MONGODB_RATING_COLLECTION)
      .format("com.mongodb.spark.sql")
      .load()
      .as[Rating] //做类型相关的自动转换  dataset，数据集
      .toDF() //alt+回车，看变量类型:ratingDF: DataFrame

    //创建一张名叫ratings的表
    ratingDF.createOrReplaceTempView("ratings") //临时视图
    //TODO: 不同的统计推荐结果
    //统计所有历史数据中每个商品的评分数
    //数据结构 -》  productId,count
    val rateMoreProductsDF = spark.sql("select productId, count(productId) as count from ratings group by productId ")

    rateMoreProductsDF
      .write
      .option("uri",mongoConfig.uri)
      .option("collection",RATE_MORE_PRODUCTS)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()

    //统计以月为单位拟每个商品的评分数
    //数据结构 -》 productId,count,time

    //创建一个日期格式化工具
    val simpleDateFormat = new SimpleDateFormat("yyyyMM")

    //UDF:user define 用户自定义
    //注册一个UDF函数，用于将timestamp装换成年月格式   1260759144000  => 201605
    spark.udf.register("changeDate",(x:Int) => simpleDateFormat.format(new Date(x * 1000L)).toInt) //10位时间戳，表示秒；13位表示毫秒

    // 将原来的Rating数据集中的时间转换成年月的格式
    val ratingOfYearMonth = spark.sql("select productId, score, changeDate(timestamp) as yearmonth from ratings")

    // 将新的数据集注册成为一张表
    ratingOfYearMonth.createOrReplaceTempView("ratingOfMonth")

    val rateMoreRecentlyProducts = spark.sql("select productId, count(productId) as count ,yearmonth from ratingOfMonth group by yearmonth,productId order by yearmonth desc, count desc")

    rateMoreRecentlyProducts
      .write
      .option("uri",mongoConfig.uri)
      .option("collection",RATE_MORE_RECENTLY_PRODUCTS)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()

    //统计每个商品的平均评分
    val averageProductsDF = spark.sql("select productId, avg(score) as avg from ratings group by productId ")

    averageProductsDF
      .write
      .option("uri",mongoConfig.uri)
      .option("collection",AVERAGE_PRODUCTS)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()

    spark.stop()
  }
}